package com.own.component.elastic.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import com.own.component.common.base.entity.po.BaseBean;
import com.own.component.elastic.entity.BaseElasticPo;
import com.own.component.elastic.entity.BaseElasticQuery;
import com.own.component.elastic.entity.BaseElasticVo;
import com.own.component.elastic.exception.ElasticException;
import com.own.component.elastic.exception.ElasticExceptionEnum;
import com.own.component.elastic.service.BaseElasticLogService;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * ABaseElasticService
 *
 * @author chenxueli
 * @date 2022-09-03 18:42:00
 */
@Slf4j
@Service
public abstract class AbstractBaseElasticLogService<
        PO extends BaseBean,
        ELASTIC_PO extends BaseElasticPo<PO>,
        ELASTIC_VO extends BaseElasticVo<ELASTIC_PO>,
        QUERY extends BaseElasticQuery
        >
        extends AbstractBaseElasticLogSearchService<PO, ELASTIC_PO, ELASTIC_VO, QUERY>
        implements BaseElasticLogService<PO, ELASTIC_PO, ELASTIC_VO, QUERY> {

    private static final Set<String> INDEX_SET = new HashSet<>();

    @Resource
    private ElasticsearchClient elasticsearchClient;

    /**
     * 创建相关索引
     *
     * @param po 数据
     */
    @Override
    public void createIndex(ELASTIC_PO po) {
        try {
            elasticsearchClient.indices().create(builder ->
                    builder.index(updateIndex(po)).mappings(mappings -> mappings.properties(mapping()))
            );
        } catch (IOException e) {
            throw new ElasticException(ElasticExceptionEnum.INDEX_CREATE_ERROR, e);
        }
    }

    /**
     * 删除相关索引
     *
     * @param po 数据
     */
    @Override
    public void deleteIndex(ELASTIC_PO po) {
        if (!checkIndex(po)) {
            return;
        }
        try {
            elasticsearchClient.indices().delete(builder -> builder.index(updateIndex(po)));
        } catch (IOException e) {
            throw new ElasticException(ElasticExceptionEnum.INDEX_DELETE_ERROR, e);
        }
    }

    /**
     * 验证相关索引
     *
     * @param po 数据
     * @return 是否存在
     */
    @Override
    public boolean checkIndex(ELASTIC_PO po) {
        var index = updateIndex(po);
        if (INDEX_SET.contains(index)) {
            return true;
        }
        try {
            var response = elasticsearchClient.indices().exists(builder -> builder.index(index));
            var flag = response.value();
            if (flag) {
                INDEX_SET.add(index);
            }
            return flag;
        } catch (IOException e) {
            throw new ElasticException(ElasticExceptionEnum.INDEX_EXIST_ERROR, e);
        }
    }

    /**
     * 插入数据
     *
     * @param po 添加的对象
     */
    @Override
    public final void add(ELASTIC_PO po) {
        if (!checkIndex(po)) {
            createIndex(po);
        }
        try {
            elasticsearchClient.index(document -> document
                    .index(updateIndex(po))
                    .routing(routing(po))
                    .id(String.valueOf(po.getId()))
                    .document(po)
            );
        } catch (IOException e) {
            throw new ElasticException(ElasticExceptionEnum.DATA_SAVE_ERROR, e);
        }
    }

    /**
     * 批量插入数据
     *
     * @param list 列表数据
     */
    @Override
    public final void batchAdd(Collection<ELASTIC_PO> list) {
        var itemList = list.stream().map(item -> new TempBulkOperation<>(
                item,
                BulkOperation.of(bulk ->
                        bulk.index(document -> document
                                .id(String.valueOf(item.getId()))
                                .document(item)
                                .routing(routing(item))
                        )
                ))
        ).toList();
        bulk(itemList);
    }

    /**
     * 修改数据
     *
     * @param id 主键id
     * @param po 添加的对象
     */
    @Override
    public final void update(String id, ELASTIC_PO po) {
        try {
            elasticsearchClient.update(document -> document.index(updateIndex(po)).id(id).doc(po), getElasticPoClass());
        } catch (IOException e) {
            throw new ElasticException(ElasticExceptionEnum.DATA_UPDATE_ERROR, e);
        }
    }

    /**
     * 修改或插入数据
     *
     * @param id 主键id
     * @param po 添加的对象
     */
    @Override
    public void upsert(String id, ELASTIC_PO po) {
        try {
            elasticsearchClient.update(document -> document.docAsUpsert(true).index(updateIndex(po)).id(id).doc(po).detectNoop(true), getElasticPoClass());
        } catch (IOException e) {
            throw new ElasticException(ElasticExceptionEnum.DATA_UPDATE_ERROR, e);
        }
    }

    /**
     * 批量修改或插入数据
     *
     * @param list 列表数据
     */
    @Override
    public void batchUpsert(Collection<ELASTIC_PO> list) {
        var itemList = list.stream().map(item -> new TempBulkOperation<>(
                item,
                BulkOperation.of(bulk ->
                        bulk.update(document ->
                                document.id(String.valueOf(item.getId())).action(a -> a.docAsUpsert(true).doc(item))
                        )
                ))).toList();
        bulk(itemList);
    }

    /**
     * 删除数据
     *
     * @param id 主键id
     */
    @Override
    public final void delete(String id) {
        try {
            // 查询数据
            var po = getPoById(id);
            if (po != null) {
                elasticsearchClient.delete(document -> document.index(updateIndex(po)).id(id));
            }
        } catch (IOException e) {
            throw new ElasticException(ElasticExceptionEnum.DATA_DELETE_ERROR, e);
        }
    }

    /**
     * 批量删除数据
     *
     * @param idList 主键id列表
     */
    @Override
    public final void batchDelete(Collection<String> idList) {
        // 查询已有的数据信息
        var list = listPoByIdList(idList);
        var itemList = list.stream().map(item -> new TempBulkOperation<>(
                item,
                BulkOperation.of(
                        bulk -> bulk.delete(document -> document.id(item.getId()))
                ))
        ).toList();
        bulk(itemList);
    }

    /**
     * 批量执行方法
     *
     * @param list 列表数据
     */
    private void bulk(Collection<TempBulkOperation<PO, ELASTIC_PO>> list) {
        // 按照索引进行分类
        var map = list.stream().collect(Collectors.groupingBy(item -> updateIndex(item.getPo())));
        // 分类执行
        map.forEach((index, values) -> {
            try {
                var ops = values.stream().map(TempBulkOperation::getOperation).toList();
                var response = elasticsearchClient.bulk(x -> x.index(index).operations(ops));
                if (response.errors()) {
                    throw new ElasticException(ElasticExceptionEnum.DATA_SAVE_ERROR);
                }
            } catch (IOException e) {
                throw new ElasticException(ElasticExceptionEnum.DATA_SAVE_ERROR, e);
            }
        });
    }

    @Getter
    @AllArgsConstructor
    public static class TempBulkOperation<
            PO extends BaseBean,
            ELASTIC_PO extends BaseElasticPo<PO>
            > {

        private ELASTIC_PO po;

        private BulkOperation operation;

    }

}
